package com.dingwen.treasure.business.service.impl;

import cn.hutool.core.convert.Convert;
import cn.hutool.json.JSONUtil;
import com.dingwen.treasure.base.constant.RabbitmqConstant;
import com.dingwen.treasure.base.enums.MsgState;
import com.dingwen.treasure.business.service.DisinfectServerService;
import com.dingwen.treasure.mybatisplus.service.TreasureMessageService;
import com.rabbitmq.client.Channel;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

/**
 * 　DisinfectServerServiceImpl
 * 　@author dingwen
 * 　@date 2022/8/10
 */
@Service
@RabbitListener(queues = {RabbitmqConstant.QUEUE_DISINFECT_SERVER}, concurrency = "10", ackMode = "MANUAL")
@Slf4j
public class DisinfectServerServiceImpl implements DisinfectServerService {

    /**
     * messageService
     */
    @Resource
    private TreasureMessageService messageService;

    /**
     * 订阅
     */
    @Override
    @Async(value = "asyncExecutor")
    @RabbitHandler
    @SneakyThrows
    public void subscribe(Message messages, String message, Channel channel) {
        long deliveryTag = messages.getMessageProperties().getDeliveryTag();
        String messageId = messages.getMessageProperties().getHeader(RabbitmqConstant.SPRING_RETURNED_MESSAGE_CORRELATION);
        MsgState mgState = MsgState.CONSUMED;
        if (!JSONUtil.isTypeJSON(message)) {
            log.error("{}消息消费，消息内容，不是JSON", RabbitmqConstant.QUEUE_OPERATOR_LOG);
            // 启用delivery tag,不进入死信信队列
            channel.basicNack(deliveryTag, Boolean.FALSE, Boolean.FALSE);
            // 修改消息状态
            mgState = MsgState.FAIL;
        }
        log.info("执行消息消费,消息ID:{}", messageId);
        messageService.updateMessageState(mgState, Convert.toLong(messageId));
        channel.basicAck(deliveryTag,Boolean.FALSE);
    }
}
